package com.thlws.study.redismq.internal.helper;

import cn.hutool.core.convert.Convert;
import com.thlws.study.redismq.context.RedisContext;
import com.thlws.study.redismq.context.RedisMessageChannel;
import com.thlws.study.redismq.internal.handler.MessageHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author HanleyTang 2019-07-31
 */
@Component
public class RedisMqHelper {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final RedisMessageListenerContainer container;

    @Resource(name = "myRedisTemplate")
    private RedisTemplate<String,Object> redisTemplate;

    public RedisMqHelper(RedisMessageListenerContainer container) {
        this.container = container;
    }


    public void subscribe(String channel, MessageHandler receiver) {

        if (RedisContext.hasChannel(channel)){
            logger.warn("已订阅该频道,channel={}",channel);
            return;
        }

        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(receiver);
        listenerAdapter.afterPropertiesSet();

        Topic topic  = new PatternTopic(channel);
        RedisMessageChannel messageChannel = new RedisMessageChannel(listenerAdapter,topic);
        RedisContext.bindChannel(channel,messageChannel);

        container.addMessageListener(listenerAdapter,topic);
    }


    public void unsubscribe(String channel) {
        if (RedisContext.hasChannel(channel)){
            RedisMessageChannel messageChannel = RedisContext.acquireChannel(channel);
            MessageListener messageListener = messageChannel.getListener();
            Topic topic  = messageChannel.getTopic();
            container.removeMessageListener(messageListener,topic);
            RedisContext.unbindChannel(channel);
        }else {
            logger.warn("未订阅该频道,channel={}",channel);
        }
    }

    public void set(String key,Object value) {
        redisTemplate.opsForValue().set(key,value);
    }

    public <T> T get(String key, Class<T> cls) {
        Object obj = get(key);
        return Convert.convert(cls, obj);
    }

    public Object get(String key) {
        return redisTemplate.opsForValue().get(key);
    }

}
